package sync

import (
	"strings"
	"sync"
	"sync/atomic"

	"gddgame.cc/galaxy/asteroid"
	"gddgame.cc/galaxy/asteroid/cluster"
	"gddgame.cc/galaxy/binocle"
	"gddgame.cc/galaxy/common/feature/orbit"
	"gddgame.cc/galaxy/satellite/formula/comet"
	"gddgame.cc/galaxy/utils"
	"gddgame.cc/galaxy/utils/def"
	"gddgame.cc/galaxy/utils/serialize"
	"gddgame.cc/galaxy/utils/uuid"
)

const (
	NumberClusterEvent      = 128
	NewBlockClusterEvent    = 129
	RemoveBlockClusterEvent = 130
	ChangeBlockClusterEvent = 131
	JoinBlockClusterEvent   = 132
	LeaveBlockClusterEvent  = 133
	LocationClusterEvent    = 134
	BindClientClusterEvent  = 135
	ClientBindClusterEvent  = 136
)

type clusterCoreInstance struct {
	Area         string
	Number       uint32            `json:"number" msg:"number"`
	Proxys       map[string]uint32 `json:"proxy" msg:"proxy"`
	Blocks       map[uint32]uint32 `json:"blocks" msg:"blocks"`
	BucketStatus map[uint32]BucketStatus
	snowFlake    *uuid.SnowFlake
	Block        *blockInstance

	Node cluster.Node

	UpperLimit  uint32 // 上限人数阀门
	LowerLimit  uint32 // 下限人数阀门
	MiddleLimit uint32 // 中间阀门
}

func (block *clusterCoreInstance) Join(member *memberInstance, node cluster.Node) {

}
func (block *clusterCoreInstance) Leave(member *memberInstance, node cluster.Node) {

}
func (block *clusterCoreInstance) UniqueId() uint32 {
	return utils.ToUint32(block.snowFlake.GenerateId())
}

func (block *clusterCoreInstance) Data(data ...interface{}) {

}

func (block *clusterCoreInstance) Location(openId string, location []byte) {

}

func (block *clusterCoreInstance) UpdateBucket(service *ClusterModeService) uint32 {
	target := uint32(0)
	blockNumber := uint32(len(block.Blocks))
	ii := make([]bool, blockNumber)
	// 获取迁移目标
	for bucket, number := range block.Blocks {
		if bucket < blockNumber {
			ii[bucket-1] = true
		}
		if number < block.UpperLimit {
			target = bucket
			break
		}
	}
	if target == 0 {
		selectBucket := uint32(0)
		for index, b := range ii {
			if b {
				continue
			}
			selectBucket = uint32(index) + 1
		}
		if selectBucket == 0 {
			selectBucket = uint32(len(ii)) + 1
		}
		target = selectBucket
	}
	if _, ok := block.Blocks[target]; !ok {
		go func() {
			// 初始化第一个bucket
			blockId := service.GetSubNodeId(block.Area, block.Block.Block, target)
			service.GetWithNodeId(block.Node.Container(), blockId)
		}()
	}
	return target
}

type BucketStatus byte

const (
	NormalBucketStatus = iota
	MaxBucketStatus
	StopBucketStatus
)

type memberProxy struct {
	OpenId    string
	Direction orbit.Direction
	Current   uint32
	Client    *cluster.Client
	Socket    *comet.Socket
}
type clusterProxyInstance struct {
	Area         string
	Number       uint32 `json:"number" msg:"number"`
	Current      uint32 `json:"current" msg:"current"`
	Window       int64  `json:"window" msg:"window"`
	Members      sync.Map
	Buckets      sync.Map
	BucketNumber uint32
	Block        *orbit.BlockLocation
	Core         *cluster.Client
	Node         cluster.Node
	cluster.Socket
}

func (proxy *clusterProxyInstance) Join(mm *memberProxy, service *ClusterModeService) (uint32, uint32) {
	//fmt.Println("join", mm.OpenId)
	number := uint32(0)
	cc := service.Block.GetSocketClient(mm.Socket)
	proxy.Buckets.Range(func(key, value interface{}) bool {
		bucket := utils.ToUint32(key)
		// 连接所有bucket
		client := service.GetSubBlock(cc, proxy.Node.Container(), proxy.Block, bucket)
		client.Join(mm.Socket, cc.OpenId)
		if bucket != proxy.Current {
			return true
		}
		mm.Current = bucket
		cc.CurrentClient = client
		number = value.(uint32) + 1
		proxy.Buckets.Store(key, number)
		return true
	})
	return mm.Current, number
}

func (proxy *clusterProxyInstance) Leave(mm *memberProxy, service *ClusterModeService) (uint32, uint32) {
	//fmt.Println("leave", mm.OpenId)
	number := uint32(0)
	cc := service.Block.GetSocketClient(mm.Socket)
	proxy.Buckets.Range(func(key, value interface{}) bool {
		bucket := utils.ToUint32(key)
		// 断开所有bucket
		// 加入、离开对应的层级
		client := service.GetSubBlock(cc, proxy.Node.Container(), proxy.Block, bucket)
		client.Leave(mm.Socket, cc.OpenId)
		if bucket != mm.Current {
			return true
		}
		number = value.(uint32) - 1
		proxy.Buckets.Store(key, number)
		return true
	})
	bucket := mm.Current
	mm.Current = 0
	cc.CurrentClient = nil
	return bucket, number
}

func (proxy *clusterProxyInstance) Remove(bucket uint32, service *ClusterModeService) {
	//fmt.Println("remove", bucket)
	blockId := service.GetSubNodeId(proxy.Area, proxy.Block, bucket)
	client := service.GetWithNodeId(proxy.Node.Container(), blockId)
	proxy.Members.Range(func(key, value interface{}) bool {
		mm := value.(*memberProxy)
		cc := service.Block.GetSocketClient(mm.Socket)
		// 断开所有bucket
		// 加入、离开对应的层级
		client.Leave(mm.Socket, cc.OpenId)
		return true
	})
}

func (proxy *clusterProxyInstance) New(bucket uint32, service *ClusterModeService) {
	//fmt.Println("new", bucket)
	blockId := service.GetSubNodeId(proxy.Area, proxy.Block, bucket)
	client := service.GetWithNodeId(proxy.Node.Container(), blockId)
	proxy.Members.Range(func(key, value interface{}) bool {
		mm := value.(*memberProxy)
		if mm.Current == bucket {
			return true
		}
		cc := service.Block.GetSocketClient(mm.Socket)
		// 加入
		client.Join(mm.Socket, cc.OpenId)
		if mm.Current == 0 {
			mm.Current = bucket
		}
		return true
	})
}

func (proxy *clusterProxyInstance) Change(bucket uint32, target uint32, service *ClusterModeService) {
	//fmt.Println("change", bucket, target)
	proxy.Members.Range(func(key, value interface{}) bool {
		mm := value.(*memberProxy)
		if mm.Current != bucket {
			return true
		}
		mm.Current = target
		cc := service.Block.GetSocketClient(mm.Socket)
		client := service.GetSubBlock(cc, proxy.Node.Container(), proxy.Block, bucket)
		cc.CurrentClient = client
		var v interface{}
		var ok bool
		v, ok = proxy.Buckets.Load(bucket)
		if ok {
			proxy.Buckets.Store(bucket, v.(uint32)-1)
		}
		v, _ = proxy.Buckets.Load(target)
		proxy.Buckets.Store(target, v.(uint32)+1)
		return true
	})
}

type clusterBlockInstance struct {
	Area      string
	Bucket    uint32
	Block     *blockInstance
	snowFlake *uuid.SnowFlake
	Core      *cluster.Client
	cluster.Node
	cluster.Socket
}

func (block *clusterBlockInstance) Join(member *memberInstance, node cluster.Node) {
	//fmt.Println("join")
	// 通知core
	//block.Core.Emit().Origin(block).List(block.Bucket, block.Block.MemberNum).Send(JoinBlockClusterEvent)
}
func (block *clusterBlockInstance) Leave(member *memberInstance, node cluster.Node) {
	//fmt.Println("leave")
	//block.Core.Emit().Origin(block).List(block.Bucket, block.Block.MemberNum).Send(LeaveBlockClusterEvent)
	// 等待服务端发送关闭
	if block.Block.Empty() {
		node.Debugf("Sync Leave->Close:%s", node.Token())
		block.Core.Leave(block, utils.ToString(block.Bucket))

		_ = node.Close()
	}
}
func (block *clusterBlockInstance) UniqueId() uint32 {
	return utils.ToUint32(block.snowFlake.GenerateId())
}

func (block *clusterBlockInstance) Data(data ...interface{}) {

}

func (block *clusterBlockInstance) Location(openId string, location []byte) {

}

type ClusterModeService struct {
	def.Logger
	binocle.Service

	Block  *BlockService
	Proxys sync.Map

	UpperLimit uint32 // 上限人数阀门
	LowerLimit uint32 // 下限人数阀门
}

func (service *ClusterModeService) ID() string {
	return "syncCluster"
}
func (service *ClusterModeService) Master(node cluster.Node) {

}
func (service *ClusterModeService) Init(node cluster.Node) {
	token := node.Token()
	node.SetConfig(cluster.ThreadSafe, true)

	index := strings.Index(token, "|")
	if index >= 0 && token[index] == '|' {
		blockIndex := strings.Index(token, "/")
		snowFlake := uuid.NewSnowFlakeSecond32(1, 3, "")
		block := orbit.LocationCM.NodeBlock(token[blockIndex+1 : index])
		coreClient := service.GetCore(node.Container(), token[:blockIndex], block)
		instance := &clusterBlockInstance{
			Area:      token[:blockIndex],
			Node:      node,
			Bucket:    utils.ToUint32(token[index+1:]),
			Block:     NewBlockInstance(token[:blockIndex], block, service, true),
			snowFlake: snowFlake,
			Core:      coreClient,
		}
		instance.Socket = cluster.NewSocket(token, instance)

		node.SetConfig(cluster.Replicate, true)
		node.SetInstance(instance)
		service.Block.InitService(node, instance.Block, instance)

		coreClient.Join(instance, utils.ToString(instance.Bucket))
		return
	}

	index = strings.Index(token, "~")
	if index >= 0 && token[index] == '~' {
		blockIndex := strings.Index(token, "/")
		block := orbit.LocationCM.NodeBlock(token[blockIndex+1 : index])
		coreClient := service.GetCore(node.Container(), token[:blockIndex], block)
		instance := &clusterProxyInstance{
			Area:    token[:blockIndex],
			Number:  0,
			Members: sync.Map{},
			Buckets: sync.Map{},
			Current: 0,
			Window:  0,
			Node:    node,
			Block:   block,
			Core:    coreClient,
		}

		instance.Socket = cluster.NewSocket(token, instance)

		// 注册进service，proxy是本地服务，可以在client中被调用
		service.Proxys.Store(token, instance)
		node.SetConfig(cluster.Replicate, true)
		node.SetInstance(instance)
		service.InitProxy(node, instance)

		coreClient.Join(instance, node.CurrentNode())
		return
	}

	index = strings.Index(token, "/")
	if index >= 0 && token[index] == '/' {
		snowFlake := uuid.NewSnowFlakeSecond32(1, 3, "")
		block := orbit.LocationCM.NodeBlock(token[index+1:])
		instance := &clusterCoreInstance{
			Area:         token[:index],
			Block:        NewBlockInstance(token[:index], block, service, false),
			snowFlake:    snowFlake,
			Proxys:       map[string]uint32{},
			Blocks:       map[uint32]uint32{},
			BucketStatus: map[uint32]BucketStatus{},
			Node:         node,
			UpperLimit:   service.UpperLimit,
			LowerLimit:   service.LowerLimit,
			MiddleLimit:  (service.UpperLimit + service.LowerLimit) / 2,
		}

		node.SetConfig(cluster.Replicate, true)
		node.SetInstance(instance)
		service.Block.InitService(node, instance.Block, instance)
		service.InitCore(node, instance)
		return
	}
}

func (service *ClusterModeService) InitCore(node cluster.Node, instance *clusterCoreInstance) {
	node.On(cluster.JoinEvent, func(message asteroid.Message) {
		token := message.GetOrigin()
		if strings.Contains(token, "|") {
			bucket := utils.ToUint32(message.Data())
			message.Response(true)
			instance.Blocks[bucket] = 0
			instance.BucketStatus[bucket] = NormalBucketStatus
			// 广播给所有的代理节点
			node.Broadcast().Data(bucket).Send(NewBlockClusterEvent)
		} else if strings.Contains(token, "~") {
			n := utils.ToString(message.Data())
			// 先回复，完成join，在broadcast
			message.Response(true)
			instance.Proxys[n] = 0
			bucket := instance.UpdateBucket(service)
			// 广播给所有的代理节点
			node.Broadcast().List(bucket, NormalBucketStatus, bucket).Send(ChangeBlockClusterEvent)
		}
	})
	node.On(cluster.LeaveEvent, func(message asteroid.Message) {
		token := message.GetOrigin()

		if strings.Contains(token, "|") {
			bucket := utils.ToUint32(message.Data())
			if _, ok := instance.Blocks[bucket]; ok {
				delete(instance.Blocks, bucket)
				delete(instance.BucketStatus, bucket)
				// 广播给所有的代理节点
				node.Broadcast().List(bucket, instance.UpdateBucket(service)).Send(RemoveBlockClusterEvent)
			}
		} else if strings.Contains(token, "~") {
			n := utils.ToString(message.Data())
			if _, ok := instance.Proxys[n]; ok {
				delete(instance.Proxys, n)
			}
		}
		message.Response(true)
		// 节点数据为空
		if len(instance.Proxys) == 0 && instance.Block.Empty() {
			_ = node.Close()
		}
	})
	// 提交数量
	node.On(NumberClusterEvent, func(message asteroid.Message) {
		bag := serialize.GetBag()
		bag.SetList(message.List())
		number := bag.ReadUint32(0)
		bag.Release()
		token := message.GetOrigin()
		diff := number
		if v, ok := instance.Proxys[token]; ok {
			diff = number - v
		}
		instance.Proxys[token] = number
		atomic.StoreUint32(&instance.Number, instance.Number+diff)
		message.Response(true)
	})
	node.On(JoinBlockClusterEvent, func(message asteroid.Message) {
		bag := serialize.GetBag()
		bag.SetList(message.List())
		bucket := bag.ReadUint32(0)
		instance.Blocks[bucket] = utils.ToUint32(bag.ReadUint32(1))
		if instance.Blocks[bucket] > instance.UpperLimit && instance.BucketStatus[bucket] != MaxBucketStatus {
			// 如果超出上限，通知
			instance.BucketStatus[bucket] = MaxBucketStatus
			node.Broadcast().List(bucket, MaxBucketStatus, instance.UpdateBucket(service)).Send(ChangeBlockClusterEvent)
		}
		bag.Release()
	})
	node.On(LeaveBlockClusterEvent, func(message asteroid.Message) {
		bag := serialize.GetBag()
		bag.SetList(message.List())
		bucket := bag.ReadUint32(0)
		instance.Blocks[bucket] = utils.ToUint32(bag.ReadUint32(1))
		if instance.Blocks[bucket] < instance.LowerLimit && instance.BucketStatus[bucket] != StopBucketStatus {
			flag := true
			for b, v := range instance.Blocks {
				if b == bucket {
					continue
				}
				if v > instance.MiddleLimit {
					flag = false
					break
				}
			}
			if flag {
				instance.BucketStatus[bucket] = StopBucketStatus
				node.Broadcast().List(bucket, StopBucketStatus, instance.UpdateBucket(service)).Send(ChangeBlockClusterEvent)
			}
		} else if instance.Blocks[bucket] == instance.UpperLimit && instance.BucketStatus[bucket] != NormalBucketStatus {
			instance.BucketStatus[bucket] = NormalBucketStatus
			node.Broadcast().List(bucket, NormalBucketStatus, instance.UpdateBucket(service)).Send(ChangeBlockClusterEvent)
		}
		bag.Release()
	})
}

func (service *ClusterModeService) InitCoreClient(client *cluster.Client) {
	client.Filter(NewBlockClusterEvent, func(message asteroid.Message, member def.Member) {
		bucket := utils.ToUint32(message.Data())
		switch socket := member.(*cluster.SocketInstance).Instance().(type) {
		case *clusterProxyInstance:
			// 代理
			atomic.AddUint32(&socket.BucketNumber, 1)
			socket.Buckets.LoadOrStore(bucket, uint32(0))
			socket.New(bucket, service)
		}
	})
	client.Filter(RemoveBlockClusterEvent, func(message asteroid.Message, member def.Member) {
		list := message.List()
		bucket := utils.ToUint32(list[0])
		target := utils.ToUint32(list[1])
		switch socket := member.(*cluster.SocketInstance).Instance().(type) {
		case *clusterProxyInstance:
			// 代理
			atomic.StoreUint32(&socket.BucketNumber, socket.BucketNumber-1)
			socket.Buckets.Delete(bucket)
			socket.Remove(bucket, service)
			socket.Current = target
		}
	})
	client.Filter(ChangeBlockClusterEvent, func(message asteroid.Message, member def.Member) {
		list := message.List()
		bucket := utils.ToUint32(list[0])
		bucketStatus := BucketStatus(utils.ToByte(list[1]))
		target := utils.ToUint32(list[2])
		switch socket := member.(*cluster.SocketInstance).Instance().(type) {
		case *clusterProxyInstance:
			socket.Current = target
			socket.Buckets.LoadOrStore(target, uint32(0))
			socket.Change(bucket, target, service)
		case *clusterBlockInstance:
			switch bucketStatus {
			case StopBucketStatus:
				nodeId := service.GetSubNodeId(socket.Block.Area, socket.Block.Block, target)
				if socket.Bucket == bucket {
					// 发动迁移
					for hash, _ := range socket.Block.Items {
						socket.Node.LocalEventAsync("transfer", []interface{}{hash, nodeId}, socket.ID(), "")
					}
				}
			}
		}
	})
	client.Filter(asteroid.DefaultFilter, func(message asteroid.Message, member def.Member) {
		//fmt.Println("filter")
		switch socket := member.(*cluster.SocketInstance).Instance().(type) {
		case *clusterProxyInstance:
			// 代理
			// 转发所有默认消息
			d, _ := message.Payload()
			socket.Node.Broadcast().Data(d).Send(message.GetEvent())
		}
	})
}
func (service *ClusterModeService) InitProxy(node cluster.Node, instance *clusterProxyInstance) {
	node.On(cluster.JoinEvent, func(message asteroid.Message) {
		openId := utils.ToString(message.Data())
		if _, ok := instance.Members.Load(openId); ok {
			message.Response(false)
			return
		}
		m := &memberProxy{Direction: orbit.NoneDirection, Current: 0}
		instance.Members.Store(openId, m)

		atomic.StoreUint32(&instance.Number, instance.Number+1)
		message.Response(true)
		// 绑定
		node.Broadcast().TargetId(message.GetOrigin()).Send(BindClientClusterEvent)
	})
	node.On(cluster.LeaveEvent, func(message asteroid.Message) {
		openId := utils.ToString(message.Data())
		m, ok := instance.Members.Load(openId)
		if !ok {
			message.Response(false)
			return
		}
		instance.Members.Delete(openId)
		atomic.StoreUint32(&instance.Number, instance.Number-1)
		message.Response(true)
		instance.Core.Emit().OriginId(instance.Node.CurrentNode()).Data(instance.Number).Send(NumberClusterEvent)

		bucket, number := instance.Leave(m.(*memberProxy), service)
		instance.Core.Emit().List(bucket, number).Send(LeaveBlockClusterEvent)

		if instance.Number == 0 {
			instance.Core.Leave(instance, node.CurrentNode())
			_ = node.Close()
		}
	})
	node.On(ClientBindClusterEvent, func(message asteroid.Message) {
		openId := utils.ToString(message.Data())
		m, ok := instance.Members.Load(openId)
		if !ok {
			message.Response(false)
			return
		}
		bucket, number := instance.Join(m.(*memberProxy), service)
		instance.Core.Emit().List(bucket, number).Send(JoinBlockClusterEvent)
	})
	node.On(LocationClusterEvent, func(message asteroid.Message) {
		bag := serialize.GetBag()
		bag.SetList(message.List())
		openId := bag.ReadString(0)
		location := orbit.LocationCM.DecodeLocation(bag.ReadBytes(1))
		bag.Release()
		m, ok := instance.Members.Load(openId)
		if !ok {
			message.Response(false)
			return
		}
		member := m.(*memberProxy)
		member.Direction = instance.Block.DirectionInline(location)
	})
}
func (service *ClusterModeService) InitProxyClient(client *cluster.Client) {
	//token := client.ID()
	//index := strings.Index(token, "~")
	//blockIndex := strings.Index(token, "/")
	//block := LocationCM.NodeBlock(token[blockIndex+1 : index])

	// 从service中加载instance，proxy是本地服务，在服务创建时就记录在service中
	i, ok := service.Proxys.Load(client.ID())
	if !ok {
		client.Error("Proxy instance is empty", client.Container().ID())
	}
	instance := i.(*clusterProxyInstance)
	client.Filter(BindClientClusterEvent, func(message asteroid.Message, member def.Member) {
		//fmt.Println("filter ping start")
		socket := member.(*comet.Socket)
		cc := service.Block.GetSocketClient(socket)
		m, ok := instance.Members.Load(cc.OpenId)
		if !ok {
			return
		}
		mm := m.(*memberProxy)
		mm.Socket = socket
		client.Emit().Data(cc.OpenId).Send(ClientBindClusterEvent)
	})
}
func (service *ClusterModeService) InitClient(client *cluster.Client) {
	token := client.ID()

	if strings.Contains(token, "|") {
		service.Block.InitClient(client)
		return
	}
	if strings.Contains(token, "~") {
		service.InitProxyClient(client)
		return
	}
	if strings.Contains(token, "/") {
		service.InitCoreClient(client)
		return
	}
}

func (service *ClusterModeService) CloseInstance(node cluster.Node) {
	token := node.Token()
	if strings.Contains(token, "|") {
		instance := node.Instance().(*clusterBlockInstance)
		service.Block.CloseInstance(node, instance.Block)
		return
	}
	if strings.Contains(token, "~") {
		service.Proxys.Delete(token)
		return
	}
	if strings.Contains(token, "/") {
		instance := node.Instance().(*clusterCoreInstance)
		service.Block.CloseInstance(node, instance.Block)
		for key, _ := range instance.Blocks {
			client := service.GetWithNodeId(node.Container(), service.GetSubNodeId(instance.Area, instance.Block.Block, key))
			client.Emit().Send(cluster.CloseEvent)
		}
	}
}

func (service *ClusterModeService) CloseClient(client *cluster.Client) {

}

func (service *ClusterModeService) NeedBlock() bool {
	return true
}

func (service *ClusterModeService) GetWithLocation(c *Client, container cluster.Container, locationByte []byte) *cluster.Client {
	block := c.LocationToBlock(locationByte)
	client := service.GetWithBlock(c, container, block)
	orbit.LocationCM.ReleaseBlock(block)
	return client
}
func (service *ClusterModeService) GetWithBlock(c *Client, container cluster.Container, block *orbit.BlockLocation) *cluster.Client {
	//fmt.Println(service.ID(), area, key)
	if block == nil {
		return nil
	}
	return container.Balance().LocalService(service.ID(), service.GetNodeId(c.Area, block)+"~"+container.Region().CurrentNode())
}

func (service *ClusterModeService) GetWithNodeId(container cluster.Container, nodeId string) *cluster.Client {
	return container.Balance().GenerateClient(service.ID(), nodeId)
}

func (service *ClusterModeService) GetNodeId(area string, block *orbit.BlockLocation) string {
	return area + "/" + block.ToNode()
}

func (service *ClusterModeService) GetSubNodeId(area string, block *orbit.BlockLocation, bucket uint32) string {
	return service.GetNodeId(area, block) + "|" + utils.ToString(bucket)
}

func (service *ClusterModeService) UpdateLocation(c *Client, location *orbit.Location) {
	c.CurrentClient.Emit().List(c.OpenId, orbit.LocationCM.EncodeLocation(location)).Send(LocationClusterEvent)
}

func (service *ClusterModeService) GetSubBlock(c *Client, container cluster.Container, block *orbit.BlockLocation, bucket uint32) *cluster.Client {
	return container.Balance().GenerateClient(service.ID(), service.GetSubNodeId(c.Area, block, bucket))
}

func (service *ClusterModeService) GetCore(container cluster.Container, area string, block *orbit.BlockLocation) *cluster.Client {
	//fmt.Println(service.ID(), area, key)
	return container.Balance().GenerateClient(service.ID(), service.GetNodeId(area, block))
}
